#include "proxy/conn/tcpserver.h"

namespace proxy{

  // ---------------------------------------------------------------------------
  TcpServer::TcpServer(std::string bind_addr, uint16 bind_port, int thread_num)
    :thread_num_(thread_num),
    bind_port_(bind_port),
    bind_addr_(bind_addr),
    is_stop_(true),
    flow_instance_(new FlowInstance(true)){
    LOG(WARNING) << "Create TcpServer ... ...";
  }

  TcpServer::~TcpServer(){
    DLOG(WARNING) << "Delete TcpServer ... ...";
  }

  bool TcpServer::Start(){
    if (!is_stop_){
      DLOG(ERROR) << "The server is running, you can't run agan";
      return false;
    }
    is_stop_ = false;
    accpet_service_.reset(new boost::asio::io_service);
    io_service_work_.push_back(boost::asio::io_service::work(*accpet_service_));

    for (int i = 0; i < thread_num_; i++){
      io_service_ptr ios(new boost::asio::io_service);
      io_services_.push_back(ios);
      io_service_work_.push_back(boost::asio::io_service::work(*ios));
      thr_grp_.create_thread(boost::bind(&TcpServer::RunThread,
        shared_from_this(), ios));
    }
    accpet_server_.reset(new TcpAccpetServer(
      io_services_, *accpet_service_, bind_addr_, bind_port_));
    accpet_server_->SetSignalAcceptError(
      boost::bind(&TcpServer::OnServerError, shared_from_this(), _1, _2));
    accpet_server_->SetSignalNewConnect(
      boost::bind(&TcpServer::OnNewConnect, shared_from_this(), _1, _2));
    accpet_server_->Start();
    thr_grp_.create_thread(boost::bind(&TcpServer::RunAccpetServer,
      shared_from_this(), accpet_service_));
    return true;
  }

  void TcpServer::Stop(){
    is_stop_ = true;
    accpet_server_->Stop();
    io_service_work_.clear();
    io_services_.clear();
  }

  void TcpServer::RunAccpetServer(io_service_ptr io_service){
    while (!is_stop_){
      LOG(ERROR) << "RunAccpetServer ... ... START";
      try{
        io_service->run();
      }
      catch (std::exception& e) {
        //BOOST_ASSERT(false);
        LOG(ERROR) << e.what();
      }
      LOG(ERROR) << "RunAccpetServer ... ... END";
    }
  }

  void TcpServer::RunThread(io_service_ptr io_service){
    while (!is_stop_){
      LOG(ERROR) << "RunThread ... ... START";
      try{
        io_service->run();
      }
      catch (std::exception& e) {
        //BOOST_ASSERT(false);
        LOG(ERROR) << e.what();
      }
      LOG(ERROR) << "RunThread ... ... END";
    }
  }

  //void TcpServer::OnNewConnect(TcpAccpetServer::Ptr accpet_server,
  //  socket_ptr connect){
  //  DLOG(INFO) << "New Connect arrive";
  //}

  void TcpServer::OnServerError(TcpAccpetServer::Ptr accpet_server,
    const boost::system::error_code& err){
    DLOG(ERROR) << err.message();
    Stop();
  }
} // namespace proxy